package com.hung.trade.common.rocketmq;

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * @author hungki
 * @createTime 2018-10-08 上午 9:17
 */
public class TradeMessageListener implements MessageListenerConcurrently {
    private IMessageProcesser iMessageProcesser;
    public TradeMessageListener(IMessageProcesser iMessageProcesser){
        this.iMessageProcesser = iMessageProcesser;
    }
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        for(MessageExt messageExt:list){
            if(!iMessageProcesser.heandlerMessage(messageExt)){
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}
